package com.wz.jiangsu.consumer;

import com.wz.jiangsu.service.es.ESWazuhService;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

/**
 * @program: wz
 * @author: wangzhou
 * @create: 2023-11-28 14:23
 * @description:
 **/
@Slf4j
@Component
public class KafkaConsumerTest {


    @Autowired
    private ESWazuhService eSWazuhService;

    @Value("${kafka.topic}")
    private String testTopic;

    // 消费监听
//    @KafkaListener(topics = {"analysisd_alert_log"})
//    public void onMessage1(ConsumerRecord<?, ?> record) {
//        System.out.println("+++++++++++++++++");
//        String value = (String) record.value();
//        System.out.println(value);
//
//    }

//    @KafkaListener(topics = {"syslog"},groupId = "syslog")
//    public void onMessage0(ConsumerRecord<?, ?> record) {
//        System.out.println("采集消息给 flink");
//        String value = (String) record.value();
//        System.out.println(value);
//    }

    /**
     * {EVENT_NETWORK_ASSENT 192.168.1.11 1697016960 654f0db4-303e-4829-9707-e39ba48a2793 10-1 map[dev_ipv4:89.34.35.18 dev_mac:08:00:37:15:e6:bc oid:1.3.6.1.2.1.1.1.0 os_desc:Darwin id-est.local 8.8.0 Darwin Kernel Version 8.8.0: Fri Sep  8 17:18:57 PDT 2006; root:xnu-792.12.6.obj~1/RELEASE_PPC Power Macintosh protocol_type:snmp] {UNKNOWN UNKNOWN    }}
     * 写到kafka的数据 >>>>>>  {"data_type":"EVENT_NETWORK_ASSENT","agent_ip":"192.168.1.11","ts":1697016960,"uuid":"654f0db4-303e-4829-9707-e39ba48a2793","seq":"10-1","payload":{"dev_ipv4":"89.34.35.18","dev_mac":"08:00:37:15:","oid":"1.3.6.1.2.1.1.1.0","os_desc":"Darwin id-est.local 8.8.0 Darwin Kernel Version 8.8.0: Fri Sep  8 17:18:57 PDT 2006; root:xnu-792.12.6.obj~1/RELEASE_PPC Power Macintosh","protocol_type":"snmp"},"assent_engine":{"dev_brand_code":"UNKNOWN","dev_brand_name":"UNKNOWN","dev_category":"","dev_model":"","friendly_name":"UNKNOWN macOS","os":"macOS"}}
     * wazuh  吐出来的内容
     *
     * @param record
     */
    @KafkaListener(topics = {"analysisd_alert_json"})
    public void onMessage2(ConsumerRecord<?, ?> record) {
        System.out.println("wauzh 吐出来的消息");
        String value = (String) record.value();
        System.out.println(value);
    }

//    @KafkaListener(topics = {"device_fingerprint"})
//    public void onMoney(ConsumerRecord<?, ?> record) {
//        System.out.println("资产识别接收到的信息");
//        String value = (String) record.value();
//        System.out.println(value);
//    }

//    @KafkaListener(topics = {"device_fingerprint_result_test"})
//    public void onMoneyReceived(ConsumerRecord<?, ?> record) {
//        System.out.println("资产识别吐出消息");
//        String value = (String) record.value();
//        System.out.println(value);
//    }

//    @KafkaListener(topics = {"device_fingerprint_result"})
//    public void onMoneyReceived_go(ConsumerRecord<?, ?> record) {
//        System.out.println("资产识别吐出消息");
//        String value = (String) record.value();
//        System.out.println(value);
//    }


    /**
     * wazuh  吐出来的内容
     *
     * @param record
     */
//    @KafkaListener(topics = {"syslog"})
//    public void onMessage21(ConsumerRecord<?, ?> record) {
////        System.out.println("态势接受到的消息");
//        String value = (String) record.value();
//        JSONObject parseObj = null;
//        try {
//            parseObj = JSONUtil.parseObj(value);
//        } catch (Exception e) {
//            log.info("消息序列化失败:{}",value);
//            return;
//        }
//        String wazuhEngine = parseObj.getStr("wazuh_engine");
//        String assentEngine = parseObj.getStr("assent_engine");
//        String cveAgent = parseObj.getStr("cve_agent");
//
//        if (StrUtil.isNotBlank(wazuhEngine)) {
////            System.out.println("wazuh吐出来的消息>>>>" + value);
//            try {
//                System.out.println("wazuh吐出来的消息>>>>" + value);
//                WazuhSecurityEventDataCommonVo bean = JSONUtil.toBean(value, WazuhSecurityEventDataCommonVo.class);
//                eSWazuhService.saveOrUpdateById(bean);
//            } catch (Exception e) {
//                log.info("消息序列化失败:{}",value);
//            }
//
//        } else if (StrUtil.isNotBlank(assentEngine)) {
//            System.out.println("资产引擎吐出来的消息>>>" + value);
//        } else if (StrUtil.isNotBlank(cveAgent)) {
////            System.out.println("cve吐出来的消息>>>" + value);
//        } else {
////            System.out.println("态势直接从主机上得到的数据>>>" + value);
//        }
//
//
//    }


    @KafkaListener(topics = {"testTopic"}, groupId = "aaa")
    public void onMessage3(ConsumerRecord<?, ?> record) {
        System.out.println(record + "：103机器的 Kafka");
        System.out.println("aaaa:bbb=============");
        String value = (String) record.value();
        System.out.println(value);
    }

    @KafkaListener(topics = {"testTopic"}, groupId = "bbb")
    public void onMessage4(ConsumerRecord<?, ?> record) {
        System.out.println(record + "：103机器的 Kafka");
        System.out.println("bbb:bbb=============");
        String value = (String) record.value();
        System.out.println(value);
    }
}
